package com.kongyz.server;


import com.kongyz.protocol.RpcDecoder;
import com.kongyz.protocol.RpcEncoder;
import com.kongyz.protocol.RpcRequest;
import com.kongyz.protocol.RpcResponse;
import com.kongyz.registry.ServiceRegistry;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.protostuff.Rpc;
import org.apache.commons.collections4.MapUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class RpcServer implements ApplicationContextAware,InitializingBean {

    private static  final Logger logger= LoggerFactory.getLogger(RpcServer.class);

    private String serverAddress;
    private ServiceRegistry serviceRegistry;

    private Map<String,Object> handlerMap=new HashMap<String, Object>();
    private static ThreadPoolExecutor threadPoolExecutor;

    private EventLoopGroup bossGroup=null;
    private EventLoopGroup workerGroup=null;

    public RpcServer(String serverAddress) {
        this.serverAddress = serverAddress;
    }

    public RpcServer(String serverAddress, ServiceRegistry serviceRegistry) {
        this.serverAddress = serverAddress;
        this.serviceRegistry = serviceRegistry;
    }

    /**
     * ApplicationContextAware 必须重写方法
     * 创建bean之后调用个方法，把整个容器作为参数传入该方法
     * @param applicationContext
     * @throws BeansException
     */
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        Map<String,Object> serviceBeanMap=applicationContext.getBeansWithAnnotation(RpcService.class);
        if(MapUtils.isNotEmpty(serviceBeanMap)){
            for (Object serviceBean:serviceBeanMap.values()){
                //加载通过RpcService注解的接口，保存到handlerMap里面，后面动态代理用到
                String interfaceName=serviceBean.getClass().getAnnotation(RpcService.class).value().getName();
                logger.info("加载RpcService:{}",interfaceName);
                handlerMap.put(interfaceName,serviceBean);
            }
        }
    }

    /**
     * InitializingBean  必须重写方法
     * 初始化是优先执行该方法
     * @throws Exception
     */
    @Override
    public void afterPropertiesSet() throws Exception {
        start();
    }

    /**
     * 手动加载RpcService
     * @param interfaceName
     * @param serviceBean
     * @return
     */
    public RpcServer addService(String interfaceName,Object serviceBean){
        if(!handlerMap.containsKey(interfaceName)){
            logger.info("加载Service:{}",interfaceName);
            handlerMap.put(interfaceName,serviceBean);
        }
        return this;
    }

    public void start() throws InterruptedException {
        if(bossGroup==null&&workerGroup==null){
            bossGroup=new NioEventLoopGroup();
            workerGroup=new NioEventLoopGroup();
            ServerBootstrap bootstrap=new ServerBootstrap();
            bootstrap.group(bossGroup,workerGroup).channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline()
                                    .addLast(new LengthFieldBasedFrameDecoder(65536,0,4,0,0))
                                    .addLast(new RpcDecoder(RpcRequest.class))
                                    .addLast(new RpcEncoder(RpcResponse.class))
                                    .addLast(new RpcHandler(handlerMap));
                        }
                    })
                    .option(ChannelOption.SO_BACKLOG,128)
                    .childOption(ChannelOption.SO_KEEPALIVE,true);
            String[] array=serverAddress.split(":");
            String host=array[0];
            int port=Integer.parseInt(array[1]);
            ChannelFuture future=bootstrap.bind(host,port).sync();
            //此处把服务器地址添加zk中注册
            if(serviceRegistry!=null){
                serviceRegistry.register(serverAddress);
            }

            future.channel().closeFuture().sync();


        }
    }


    public static void submit(Runnable task){
        if(threadPoolExecutor ==null){
            synchronized (RpcServer.class){
                if(threadPoolExecutor==null){
                    threadPoolExecutor=new ThreadPoolExecutor(16,16,600L,
                            TimeUnit.SECONDS,new ArrayBlockingQueue<Runnable>(65536));

                }
            }
        }
        threadPoolExecutor.submit(task);
    }



}
